package p.ithorns.framework.redis.stream;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;
import org.springframework.scheduling.annotation.Scheduled;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * RedisStreamContext
 *
 * @author ithorns
 * @version 1.0.0
 * @date 2022/12/10 12:42
 */
@Scope("singleton")
@AutoConfigureAfter(RedisStreamConfig.class)
public class RedisStreamContext implements
        ApplicationListener<ContextRefreshedEvent>, DisposableBean {

    private final static Logger log = LoggerFactory.getLogger(RedisStreamContext.class);

    /**
     * Stream Keys
     */
    private static final Set<String> STREAM_KEYS = new HashSet<>();

    /**
     * 队列最大消息保存数量
     */
    private final static int MAX_LEN = 2000;
    private final RedisStreamFactory factory;
    private final RedisStreamTemplate template;
    private final RedisStreamErrorHandler errorHandler;
    private final ConcurrentMap<Class<?>, StreamMessageListenerContainer> containerMap =
            new ConcurrentHashMap<>(16);

    public RedisStreamContext(RedisConnectionFactory connectionFactory,
                              RedisStreamTemplate template,
                              RedisStreamErrorHandler errorHandler) {
        this.template = template;
        this.factory = new RedisStreamFactory(connectionFactory, template);
        this.errorHandler = errorHandler;
        log.info("加载RedisStreamContext.");
    }

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        ApplicationContext ctx = event.getApplicationContext();
        DefaultListableBeanFactory beanFactory = (DefaultListableBeanFactory) ctx.getAutowireCapableBeanFactory();
        Map<String, RedisStreamListener> beans = ctx.getBeansOfType(RedisStreamListener.class);
        for (Map.Entry<String, RedisStreamListener> entry : beans.entrySet()) {
            RedisStreamListener listener = entry.getValue();
            final Class targetType = listener.getTargetType();

            StreamMessageListenerContainer container = containerMap.computeIfAbsent(targetType,
                    key -> factory.createContainer(targetType, null, errorHandler));

            Subscription subscription = factory.createSubscription(container, listener);

            STREAM_KEYS.add(listener.getStream());

            // 动态注册bean到Spring上下文，断线后重订阅用
            String simpleName = listener.getSimpleName();
            beanFactory.registerSingleton(simpleName, subscription);
            log.info("发起订阅: {}", simpleName);

            container.start();
        }

        beanFactory.registerSingleton("RedisContainers", containerMap.values());
    }


    @Override
    public void destroy() throws Exception {
        for (StreamMessageListenerContainer container : containerMap.values()) {
            container.stop();
        }
    }

    /**
     * 定期裁剪StreamKeys
     */
    @Scheduled(cron = "0 0 2 * * ?")
    public void scheduleTrimStreamKey() {
        final ArrayList<String> streamKeys = new ArrayList<>(STREAM_KEYS);
        template.execute(new SessionCallback() {
            @Override
            public Object execute(RedisOperations operations) throws DataAccessException {
                for (String streamKey : streamKeys) {
                    operations.opsForStream().trim(streamKey, MAX_LEN, true);
                }
                return null;
            }
        });

        log.info("Execute RedisStream Keys Xtrim.");
    }

}